//package com.example.service;
//
//import com.example.Utils.Env;
//import com.example.entity.SellAndBuy;
//import com.example.entity.StockTrade;
//import jakarta.annotation.PostConstruct;
//import jakarta.annotation.Resource;
//import lombok.extern.slf4j.Slf4j;
//import org.apache.flink.api.common.functions.AggregateFunction;
//import org.apache.flink.configuration.Configuration;
//import org.apache.flink.streaming.api.datastream.DataStream;
//import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
//import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
//import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
//import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
//import org.apache.flink.streaming.api.windowing.time.Time;
//import org.apache.flink.streaming.api.windowing.triggers.EventTimeTrigger;
//import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
//import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;
//import org.apache.flink.util.Collector;
//import org.springframework.stereotype.Component;
//import redis.clients.jedis.Jedis;
//
//@Component
//@Slf4j
//public class StockSellAndBuy {
//    @Resource private Env env_;
//    @PostConstruct
//    public void init() {
//        new Thread(() -> {
//            try {
//                startFlinkJob();
//            } catch (Exception e) {
//                log.error("Error starting Flink job", e);
//            }
//        }).start();
//    }
//
//    public  void startFlinkJob() throws Exception {
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//
//        DataStream<StockTrade> tradeStream = env_.getEnv(env);
//
//        log.info("Trade stream created and watermarks assigned");
//
//        // 1 小时窗口
//        tradeStream
//                .keyBy(StockTrade::getStockCode)
//                .window(TumblingEventTimeWindows.of(Time.hours(1)))
//                .trigger(EventTimeTrigger.create())
//                .aggregate(new TradeAggregator(), new TradeResultFunction("1hour"))
//                .addSink(new AccumulatingRedisSink(redisConfig, "1hour"));
//
//        // 当天累计窗口
////        tradeStream
////                .keyBy(StockTrade::getStockCode)
////                .window(GlobalWindows.create())
////                .aggregate(new TradeAggregator(), new TradeResultFunction("1day"))
////                .addSink(new AccumulatingRedisSink(redisConfig, "1day"));
//
//        env.execute("Stock Trade Analysis");
//    }
//
//    public FlinkJedisPoolConfig redisConfig = new FlinkJedisPoolConfig.Builder()
//            .setHost("localhost")
//            .setPort(6379)
//            .build();
//
//    // 自定义Sink函数
//    public static class AccumulatingRedisSink extends RichSinkFunction<SellAndBuy> {
//        private transient Jedis jedis;
//        private final FlinkJedisPoolConfig config;
//        private final String windowType;
//
//        public AccumulatingRedisSink(FlinkJedisPoolConfig config, String windowType) {
//            this.config = config;
//            this.windowType = windowType;
//        }
//
//        @Override
//        public void open(Configuration parameters) throws Exception {
//            super.open(parameters);
//            jedis = new Jedis(config.getHost(), config.getPort());
//        }
//
//        @Override
//        public void close() throws Exception {
//            if (jedis != null) {
//                jedis.close();
//            }
//            super.close();
//        }
//
//        @Override
//        public void invoke(SellAndBuy value, Context context) throws Exception {
//            String existingValue = jedis.get("2:");
//
//            double totalBuyAmount;
//            double totalSellAmount;
//            double totalBuyVolumeCount;
//            double totalSellVolumeCount;
//
//            if (existingValue != null) {
//                String[] parts = existingValue.split(":");
//                totalBuyAmount = Double.parseDouble(parts[0]);
//                totalSellAmount = Double.parseDouble(parts[1]);
//                totalBuyVolumeCount = Double.parseDouble(parts[2]);
//                totalSellVolumeCount = Double.parseDouble(parts[3]);
//            } else {
//                System.out.println("我是null");
//                totalBuyAmount = 0.0;
//                totalSellAmount = 0.0;
//                totalBuyVolumeCount = 0L;
//                totalSellVolumeCount = 0L;
//            }
//
//            totalBuyAmount += Double.parseDouble(value.getBuyAmount());
//            totalSellAmount += Double.parseDouble(value.getSellAmount());
//            totalBuyVolumeCount += Double.parseDouble(value.getBuyVolumeCount());
//            totalSellVolumeCount += Double.parseDouble(value.getSellVolumeCount());
////            System.out.println("Total Buy Amount: " + totalBuyAmount);
////            System.out.println("Total Sell Amount: " + totalSellAmount);
////            System.out.println("Total Buy Volume Count: " + totalBuyVolumeCount);
////            System.out.println("Total Sell Volume Count: " + totalSellVolumeCount);
//
//            jedis.set("2:", totalBuyAmount + ":" + totalSellAmount + ":" + totalBuyVolumeCount + ":" + totalSellVolumeCount);
//            final long delayInterval = 100;
//            Thread.sleep(delayInterval);
//        }
//    }
//
//    // 定义聚合函数
//    public static class TradeAggregator implements AggregateFunction<StockTrade, TradeAccumulator, TradeAccumulator> {
//        @Override
//        public TradeAccumulator createAccumulator() {
//            return new TradeAccumulator();
//        }
//
//        @Override
//        public TradeAccumulator add(StockTrade value, TradeAccumulator accumulator) {
//            if ("买入".equals(value.getTradeType())) {
//                accumulator.totalBuyAmount += value.getTradeVolume();
//                accumulator.totalBuyVolumeCount++;
//            } else if ("卖出".equals(value.getTradeType())) {
//                accumulator.totalSellAmount += value.getTradeVolume();
//                accumulator.totalSellVolumeCount++;
//            }
//
//            return accumulator;
//        }
//
//        @Override
//        public TradeAccumulator getResult(TradeAccumulator accumulator) {
//            TradeAccumulator.type++;
//            return accumulator;
//        }
//
//        @Override
//        public TradeAccumulator merge(TradeAccumulator a, TradeAccumulator b) {
//            a.totalBuyAmount += b.totalBuyAmount;
//            a.totalSellAmount += b.totalSellAmount;
//            a.totalBuyVolumeCount += b.totalBuyVolumeCount;
//            a.totalSellVolumeCount += b.totalSellVolumeCount;
//            return a;
//        }
//    }
//
//    // 定义累加器
//    public static class TradeAccumulator {
//        public static int type = 0;
//        public String stockCode;
//        public double totalBuyAmount = 0.0;
//        public double totalSellAmount = 0.0;
//        public long totalBuyVolumeCount = 0L;
//        public long totalSellVolumeCount = 0L;
//
//        public TradeAccumulator() {}
//
//        public TradeAccumulator(String stockCode) {
//            this.stockCode = stockCode;
//        }
//    }
//
//    // 定义结果函数
//    public static class TradeResultFunction implements WindowFunction<TradeAccumulator, SellAndBuy, String, TimeWindow> {
//        private final String windowType;
//
//        public TradeResultFunction(String windowType) {
//            this.windowType = windowType;
//        }
//
//        @Override
//        public void apply(String key, TimeWindow window, Iterable<TradeAccumulator> input, Collector<SellAndBuy> out) throws Exception {
//            TradeAccumulator accumulator = input.iterator().next();
//            log.info("Processing window for stock code: {}, window type: {}, window: [{} - {}], total buy amount: {}, total sell amount: {}, total buy volume count: {}, total sell volume count: {}  这是第{}次",
//                    key, windowType, window.getStart(), window.getEnd(), accumulator.totalBuyAmount, accumulator.totalSellAmount, accumulator.totalBuyVolumeCount, accumulator.totalSellVolumeCount, TradeAccumulator.type);
//            out.collect(new SellAndBuy(key, String.valueOf(accumulator.totalBuyAmount), String.valueOf(accumulator.totalSellAmount), String.valueOf(accumulator.totalBuyVolumeCount), String.valueOf(accumulator.totalSellVolumeCount)));
//        }
//    }
//
//
//
//
//}
//
//
//        // �为了在`StockSellAndBuy`类中实现当天累计的买入和卖出交易量，我们可以参考`StockTradeAnalysis`类中的代码结构，但需要进行一些调整以适应新的需求。以下是一个示例实现：
//
